#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <string.h>
#include <sqlite3.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <time.h>
#include <unistd.h>
#include "cJSON.h"
#include "shmem.h"
#include "msg_queue_peer.h"
#include "MQTTClient.h"

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define TOPICDOWN "/app/data/down"
#define TOPICUP "/app/data/up"
#define PAYLOAD
#define QOS 1
#define TIMEOUT 10000L

volatile MQTTClient_deliveryToken deliveredtoken;
struct shm_param para;
struct data_node *data = NULL; // 接收共享内存中结构体首地址
int *p = NULL;                 // 用来接收共享内存的首地址
volatile int refresh = 0;      // 刷新上报标志位
struct que_order msg;          // 消息队列结构体
sqlite3 *db;                   // 数据库句柄
char *history = NULL;          // 历史记录查询数组
volatile int query = 0;
volatile int control = 0;
int i = 0;
time_t t = 0;

void delivered(void *context, MQTTClient_deliveryToken dt);
// MQTT订阅端回调函数
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void connlost(void *context, char *cause);
// sqlite3数据库回调函数
int callback(void *, int, char **, char **);
// json序列化函数
char *GetJson(int len, struct data_node *start, struct data_node *sign);
// 判断变化上报函数
void JudgeUp(MQTTClient client, int len, MQTTClient_message pubmsg, MQTTClient_deliveryToken token, cJSON *type,
             struct data_node *begin, struct data_node *sign, cJSON *period, sqlite3 *db);

int main(int argc, char const *argv[])
{
    // 定义插入数据库变量
    struct sqlite sq;
    char buff[3072] = "";
    // 初始化共享内存
    shm_init(&para, "five", 1024);
    // 获取共享内存首地址
    p = shm_getaddr(&para);
    printf("%d\n", *p);
    // 获取共享内存结构体的首地址
    data = (struct data_node *)p + 1;
    printf("%p\n", data);
    struct data_node *sign = data;  // 结构体首地址的标记
    struct data_node *start = data; // 序列化使用的指针
    struct data_node *begin = data; // 变化上报使用的指针
    // 打开点表文件
    int fd = 0;
    fd = open("/mnt/config/five.json", O_RDWR);
    if (fd < 0)
    {
        printf("open err %d\n", __LINE__);
        return -1;
    }
    // 将文件内容读取到数组里
    if (read(fd, buff, 3072) < 0)
    {
        printf("read is not finish %d", __LINE__);
        return -1;
    }
    // 关闭文件描述符
    close(fd);
    // 反序列化
    cJSON *root = cJSON_Parse(buff);
    cJSON *report = cJSON_GetObjectItem(root, "report");
    // 获取上报模式
    cJSON *type = cJSON_GetObjectItem(report, "type");
    // 获取周期上报的间隔时间
    cJSON *period = cJSON_GetObjectItem(report, "period");
    printf("%d\n", __LINE__);
    // 计算节点数量
    cJSON *stm = cJSON_GetObjectItem(root, "stm32");
    printf("%d\n", __LINE__);
    cJSON *stm_arr = cJSON_GetObjectItem(stm, "data");
    int ls = cJSON_GetArraySize(stm_arr);

    cJSON *mod = cJSON_GetObjectItem(root, "modbus");
    cJSON *mod_arr = cJSON_GetObjectItem(mod, "data");
    int lm = cJSON_GetArraySize(mod_arr);
    int len = ls + lm;
    // 将结构体数量写入共享内存的首地址中
    *p = len;
    printf("%d\n", *p);
    // 将数组中的内容反序列化
    // stm32共享内存的结构体
    cJSON *stm_data = cJSON_GetObjectItem(stm, "data");
    for (int i = 0; i < ls; i++)
    {
        cJSON *node = cJSON_GetArrayItem(stm_data, i);
        cJSON *key = cJSON_GetObjectItem(node, "key");
        data->key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(node, "type");
        data->type = type->valueint;
        data->dev_type = 3;
        data->ret = -1;
        data++;
    }
    // modbus共享内存的结构体
    cJSON *mod_data = cJSON_GetObjectItem(mod, "data");
    for (int i = 0; i < lm; i++)
    {
        cJSON *node = cJSON_GetArrayItem(mod_data, i);
        cJSON *key = cJSON_GetObjectItem(node, "key");
        data->key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(node, "type");
        data->type = type->valueint;
        data->dev_type = 1;
        data->ret = -1;
        data++;
    }
    // 打开数据库
    if (sqlite3_open("./history.db", &db) != SQLITE_OK)
    {
        // stderr:标准错误
        fprintf(stderr, "%d open err:%s\n", __LINE__, sqlite3_errmsg(db));
        return -1;
    }
    // 数据库操作
    char *errmsg = NULL;
    // 创建刷新上报的表
    if (sqlite3_exec(db, "create table if not exists history(time time_t ,temp float,humid float)", NULL, NULL, &errmsg))
    {
        fprintf(stderr, "%d create err:%s\n", __LINE__, errmsg);
        return -1;
    }
    // 清空刷新上报的表
    if (sqlite3_exec(db, "DELETE FROM history", NULL, NULL, &errmsg))
    {
        fprintf(stderr, "create err:%s\n", errmsg);
        return -1;
    }
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    // 创建一个客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    // 设置回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    // 连接服务器
    if (MQTTClient_connect(client, &conn_opts) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect%d\n", __LINE__);
        exit(EXIT_FAILURE);
    }
    // 订阅主题
    MQTTClient_subscribe(client, TOPICDOWN, QOS);
    // int i = 0;
    do
    {
        // 判断上报模式  0:刷新上报 1:变化上报 2:周期上报
        JudgeUp(client, len, pubmsg, token, type, begin, sign, period, db);
        // 历史查询
        if (query == 1)
        {
            pubmsg.payload = history;
            pubmsg.payloadlen = (int)strlen(history);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            printf("历史查询回复:%s\n", (char *)(pubmsg.payload));
            MQTTClient_publishMessage(client, TOPICUP, &pubmsg, &token);
            MQTTClient_waitForCompletion(client, token, TIMEOUT);
            free(history);
            query = 0;
        }
    } while (1);
    MQTTClient_unsubscribe(client, TOPICDOWN);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return 0;
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    deliveredtoken = dt;
}
// 回调函数
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    time_t dd;
    control = 1;
    // 文件描述符
    char *payloadptr = NULL;
    // 接收客户端发来的数据
    payloadptr = message->payload;
    if (strstr(payloadptr, "type") == NULL)
    {
        printf("please send JSON\n");
        return -1;
    }
    // 反序列化
    cJSON *root = cJSON_Parse(payloadptr);
    cJSON *type = cJSON_GetObjectItem(root, "type");
    switch (type->valueint)
    {
    case 1:
        printf("请求刷新命令:%s\n", payloadptr);
        refresh = 1;
        break;
    case 2:
    {
        printf("修改设备数据命令:%s\n", payloadptr);
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *val = cJSON_GetObjectItem(data, "val");
        if (key->valueint > 300)
        {
            msg.tag = 3;
            msg.key = key->valueint;
            strcpy(msg.val, val->valuestring);
        }
        else
        {
            msg.tag = 1;
            msg.key = key->valueint;
            strcpy(msg.val, val->valuestring);
        }
        msg_queue_send("five", &msg, sizeof(msg), 0);
    }
    break;
    case 3:
    {
        printf("修改上报命令:%s\n", payloadptr);
        char buff[3072] = "";
        // 解析发来的命令获取上报类型和上报周期
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *type = cJSON_GetObjectItem(data, "type");
        cJSON *period = cJSON_GetObjectItem(data, "period");
        // 打开点表文件
        int fd = open("/mnt/config/five.json", O_RDWR);
        if (fd < 0)
        {
            printf("open err %d\n", __LINE__);
            return -1;
        }
        // 读出点表文件
        read(fd, buff, 3072);
        // 关闭文件描述符
        close(fd);

        cJSON *changeRoot = cJSON_Parse(buff);
        cJSON *changeReport = cJSON_GetObjectItem(changeRoot, "report");
        // 替换上报模式节点
        cJSON_ReplaceItemInObject(changeReport, "type", cJSON_CreateNumber(type->valueint));
        // 替换上报周期
        cJSON_ReplaceItemInObject(changeReport, "period", cJSON_CreateNumber(period->valueint));
        char *change = cJSON_Print(changeRoot);
        // 重新打开文件并清空
        fd = open("/mnt/config/five.json", O_RDWR | O_CREAT | O_TRUNC, 0777);
        if (fd < 0)
        {
            printf("open err %d\n", __LINE__);
            return -1;
        }
        // 覆盖原来的文件
        write(fd, change, strlen(change));
        printf("%d\n", __LINE__);
        // 释放空间
        free(change);
        // 关闭文件描述符
        close(fd);
    }
    break;
    case 4:
    {
        int row = 0;
        int column = 0;
        char json[3072] = "";
        char sql[128] = "";
        char *errmsg = NULL;
        char **result = NULL;
        printf("历史查询请求:%s\n", payloadptr);
        // 解析Qt下发的json指令
        cJSON *C_root = cJSON_Parse(payloadptr);
        cJSON *data = cJSON_GetObjectItem(C_root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *limit = cJSON_GetObjectItem(data, "limit");
        cJSON *min = cJSON_GetArrayItem(limit, 0);
        int low = min->valueint;
        cJSON *max = cJSON_GetArrayItem(limit, 1);
        int high = max->valueint;
        // 读出点表,获取上报模式
        int fd = open("/mnt/config/five.json", O_RDWR);
        if (fd < 0)
        {
            printf("open err %d\n", __LINE__);
            return -1;
        }
        if (read(fd, json, 3072) < 0)
        {
            printf("read is not finish %d", __LINE__);
            return -1;
        }
        cJSON *gen = cJSON_Parse(json);
        cJSON *report = cJSON_GetObjectItem(gen, "report");
        cJSON *period = cJSON_GetObjectItem(report, "period");

        // 序列化历史查询回复
        // 创建根节点
        cJSON *root = cJSON_CreateObject();
        cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(4));
        cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
        // 向根节点中添加数组节点
        // 创建数组节点
        cJSON *arr = cJSON_CreateArray();
        printf("%d  %ld\n", __LINE__, t);
        while (t != low)
        {
            low++;
        }
        printf("%d %d\n", __LINE__, low);
        while (low <= high)
        {
            sprintf(sql, "select * from history where time=%d", low);
            if (sqlite3_get_table(db, sql, &result, &row, &column, &errmsg) == SQLITE_OK)
            {
                printf("%d\n", __LINE__);
                cJSON *child = cJSON_CreateObject();
                printf("%d\n", __LINE__);
                cJSON_AddItemToObject(child, "time", cJSON_CreateString(result[3]));
                printf("%d\n", __LINE__);
                cJSON_AddItemToObject(child, "temp", cJSON_CreateString(result[4]));
                cJSON_AddItemToObject(child, "humid", cJSON_CreateString(result[5]));
                cJSON_AddItemToArray(arr, child);
            }
            low += period->valueint;
        }
        cJSON_AddItemToObject(root, "data", arr);
        history = cJSON_Print(root);
        // printf("%s\n", history);
        // 释放空间
        cJSON_Delete(root);
        query = 1;
        close(fd);
    }
    break;
    default:
        break;
    }
    control = 0;
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

// 序列化函数
char *GetJson(int len, struct data_node *start, struct data_node *sign)
{
    char buf[8] = "";
    // 创建根节点
    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    // 向根节点中添加数组节点
    // 创建数组节点
    cJSON *arr = cJSON_CreateArray();
    for (int i = 0; i < len; i++)
    {
        // 创建数组节点的子节点
        cJSON *child = cJSON_CreateObject();
        cJSON_AddItemToObject(child, "key", cJSON_CreateNumber(start->key));
        // 判断数据的类型 1：bool类型 2：int型  3：float型
        switch (start->type)
        {
        case 1:
            sprintf(buf, "%d", start->new_val.b_val);
            cJSON_AddItemToObject(child, "val", cJSON_CreateString(buf));
            break;
        case 2:
            sprintf(buf, "%d", start->new_val.i_val);
            cJSON_AddItemToObject(child, "val", cJSON_CreateString(buf));
            break;
        case 3:
            sprintf(buf, "%.2f", start->new_val.f_val);
            cJSON_AddItemToObject(child, "val", cJSON_CreateString(buf));
            break;
        default:
            break;
        }
        cJSON_AddItemToArray(arr, child);
        start++;
    }
    start = sign;
    // 将数组节点插入到根节点
    cJSON_AddItemToObject(root, "data", arr);
    // 将json树转换成字符串
    char *p = cJSON_Print(root);
    // 释放空间
    cJSON_Delete(root);
    return p;
}

// 上报函数
void JudgeUp(MQTTClient client, int len, MQTTClient_message pubmsg, MQTTClient_deliveryToken token, cJSON *type,
             struct data_node *begin, struct data_node *sign, cJSON *period, sqlite3 *db)
{
    time_t tt;
    int flag = 0;
    char buf[8] = "";
    char sql[256] = "";
    char *errmsg = NULL;
    float temp = 0, humid = 0; // 将共享内存中的结构体序列化
                               // printf("type:%d\n", type->valueint);
    switch (type->valueint)
    {
    // 刷新上报
    case 0:
    {
        if (refresh == 1)
        {
            // 将共享内存中的结构体序列化
            char *send = GetJson(len, begin, sign);
            // 将要发送的数据给pubmsg
            pubmsg.payload = send;
            pubmsg.payloadlen = (int)strlen(send);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            printf("刷新上报回复:%s\n", (char *)(pubmsg.payload));
            MQTTClient_publishMessage(client, TOPICUP, &pubmsg, &token);
            MQTTClient_waitForCompletion(client, token, TIMEOUT);
            // 释放空间
            free(send);
            refresh = 0;
        }
    }
    break;
    // 变化上报
    case 1:
    {
        // 创建根节点
        cJSON *root = cJSON_CreateObject();
        cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
        cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
        // 创建数组节点
        cJSON *arr = cJSON_CreateArray();
        // 遍历共享内存中的结构体
        for (int i = 0; i < len; i++)
        {
            // 判断数据的类型 1：bool类型 2：int型  3：float型
            switch (begin->type)
            {
            case 1:
                if (begin->old_val.b_val != begin->new_val.b_val)
                {
                    begin->old_val.b_val = begin->new_val.b_val;
                    // 创建数组子节点
                    cJSON *child = cJSON_CreateObject();
                    cJSON_AddItemToObject(child, "key", cJSON_CreateNumber(begin->key));
                    sprintf(buf, "%d", begin->new_val.b_val);
                    cJSON_AddItemToObject(child, "val", cJSON_CreateString(buf));
                    cJSON_AddItemToArray(arr, child);
                    flag = 1;
                }
                break;
            case 2:
                if (begin->old_val.i_val != begin->new_val.i_val)
                {
                    begin->old_val.i_val = begin->new_val.i_val;
                    // 创建数组子节点
                    cJSON *child = cJSON_CreateObject();
                    cJSON_AddItemToObject(child, "key", cJSON_CreateNumber(begin->key));
                    sprintf(buf, "%d", begin->new_val.i_val);
                    cJSON_AddItemToObject(child, "val", cJSON_CreateString(buf));
                    cJSON_AddItemToArray(arr, child);
                    flag = 1;
                }
                break;
            case 3:
                if (begin->old_val.f_val != begin->new_val.f_val)
                {
                    begin->old_val.f_val = begin->new_val.f_val;
                    // 创建数组子节点
                    cJSON *child = cJSON_CreateObject();
                    cJSON_AddItemToObject(child, "key", cJSON_CreateNumber(begin->key));
                    sprintf(buf, "%.2f", begin->new_val.f_val);
                    cJSON_AddItemToObject(child, "val", cJSON_CreateString(buf));
                    cJSON_AddItemToArray(arr, child);
                    flag = 1;
                }
                break;
            default:
                break;
            }
            begin++;
        }
        begin = sign;
        // 将数组节点添加到根节点
        cJSON_AddItemToObject(root, "data", arr);
        char *buf = cJSON_Print(root);
        // sleep(1);
        pubmsg.payload = buf;
        pubmsg.payloadlen = (int)strlen(buf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        if (flag == 1)
        {
            // 将共享内存中的结构体序列化
            char *send = GetJson(len, begin, sign);
            // 将要发送的数据给pubmsg
            pubmsg.payload = send;
            pubmsg.payloadlen = (int)strlen(send);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            printf("变化上报回复:%s\n", (char *)(pubmsg.payload));
            if (control == 0)
            {
                MQTTClient_publishMessage(client, TOPICUP, &pubmsg, &token);
            }
            MQTTClient_waitForCompletion(client, token, TIMEOUT);
            flag = 0;
            // 释放空间
            free(send);
        }
        // 释放内存
        free(buf);
        cJSON_Delete(root);
    }
    break;

    // 周期上报
    case 2:
    {
        // 将共享内存中的结构体序列化
        char *send = GetJson(len, begin, sign);
        // 将要发送的数据给pubmsg
        pubmsg.payload = send;
        pubmsg.payloadlen = (int)strlen(send);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        printf("周期上报回复:%s\n", (char *)(pubmsg.payload));
        MQTTClient_publishMessage(client, TOPICUP, &pubmsg, &token);
        MQTTClient_waitForCompletion(client, token, TIMEOUT);
        // 释放内存
        free(send);
        if (i == 0)
        {
            t = time(&tt);
            i = 1;
        }
        else
        {
            time(&tt);
        }
        printf("%ld\n", tt);
        temp = begin->new_val.f_val;
        begin += 1;
        humid = begin->new_val.f_val;
        begin = sign;
        sprintf(sql, "insert into history values(%ld,%.2f,%.2f)", tt, temp, humid);
        if (sqlite3_exec(db, sql, NULL, NULL, &errmsg))
        {
            fprintf(stderr, "%d insert err:%s\n", __LINE__, errmsg);
            return;
        }
        sleep(period->valueint);
    }
    break;

    default:

        break;
    }
}
